package cn.like.stream.helper;

import cn.hutool.json.JSONUtil;
import cn.like.function.Func;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collection;

/**
 * @author <a href="mailto:likelovec@gmail.com">like</a>
 * @date 2021/7/21 0:18
 */
public final class RocketMessageCreator<T> {

    private final Message message;
    private final Charset charset = StandardCharsets.UTF_8;

    private RocketMessageCreator(String topic, T body) {
        byte[] payloads;
        try {
            if (null == body) {
                throw new RuntimeException("the message cannot be empty");
            }
            // 如果是 String 类型，则直接获得其 byte[] 内容。
            if (body instanceof String) {
                payloads = ((String) body).getBytes(charset);
                // 如果是 byte[] 类型，则直接使用即可
            } else if (body instanceof byte[]) {
                payloads = (byte[]) body;
                // 如果是复杂对象类型，则使用 MessageConverter 进行转换成字符串，然后再获得字符串的 byte[] 内容。
            } else {
                String jsonObj = JSONUtil.toJsonStr(body);
                if (null == jsonObj) {
                    throw new RuntimeException(String.format(
                            "empty after conversion [payloadClass:%s,payloadObj:%s]", body.getClass(), body));
                }
                payloads = jsonObj.getBytes(charset);
            }
        } catch (Exception e) {
            throw new RuntimeException("convert to RocketMQ message failed.", e);
        }
        this.message = new Message(topic, payloads);
    }

    /**
     * of
     *
     * @param topic 主题
     * @param body  消息体
     * @return {@link RocketMessageCreator<T>}
     */
    public static <T> RocketMessageCreator<T> of(String topic, T body) {
        return new RocketMessageCreator<T>(topic, body);
    }

    /**
     * 一个应用尽可能用一个Topic，而消息子类型则可以用tags来标识。
     * tags可以由应用自由设置，只有生产者在发送消息设置了tags，消费方在订阅消息时才可以利用tags通过broker做<h2>消息过滤</h2>
     *
     * @param tag 标签
     * @return {@link RocketMessageCreator<T>}
     */
    public RocketMessageCreator<T> tag(String tag) {
        if (Func.isNotBlack.test(tag))
            message.setTags(tag);
        return this;
    }

    /**
     * 消息发送的目的地
     *
     * @param topic 消息的主题
     * @return {@link RocketMessageCreator<T>}
     */
    public RocketMessageCreator<T> topic(String topic) {
        if (Func.isNotBlack.test(topic))
            message.setTopic(topic);
        return this;
    }

    /**
     * @param keys 业务相关的唯一表示，用于message的查找
     * @return {@link RocketMessageCreator<T>}
     */
    public RocketMessageCreator<T> keys(String keys) {
        if (Func.isNotBlack.test(keys))
            message.setKeys(keys);
        return this;
    }

    /**
     * @param keys 业务相关的唯一表示，用于message的查找
     * @return {@link RocketMessageCreator<T>}
     */
    public RocketMessageCreator<T> keys(Collection<String> keys) {
        if (Func.isNotBlack.test(keys))
            message.setKeys(keys);
        return this;
    }

    /**
     * @param delayLevel 延迟等级（mq消息中，批量发送不能指定）{@link cn.like.stream.helper.MessageDelayLevel}
     * @return {@link RocketMessageCreator<T>}
     */
    public RocketMessageCreator<T> delay(MessageDelayLevel delayLevel) {
        if (Func.isNotBlack.test(delayLevel))
            message.setDelayTimeLevel(delayLevel.level());
        return this;
    }

    public Message create() {
        return this.message;
    }
}
